package main

import (
	"bufio"
	"fmt"
	"os"
	"strconv"
	"study/gointro/pipeline"
)

func main() {
	p := createNetworkPipeline("small.in", 512, 4)
	writeToFile(p, "small.out")
	printFile("small.out")

}

func printFile(filename string) {
	file, err := os.Open(filename)
	if err != nil {
		panic(err)

	}
	defer file.Close()

	//将文件中的数据读到管道中去
	p := pipeline.ReaderSource(file, -1)

	count := 0
	for v := range p {
		fmt.Println(v)
		count++
		if count >= 100 {
			break
		}
	}

}

//将管道的数据写入文件
func writeToFile(p <-chan int, filename string) {
	file, err := os.Create(filename)
	if err != nil {
		panic(err)
	}
	defer file.Close()

	writer := bufio.NewWriter(file)
	defer writer.Flush()

	pipeline.WriterSink(writer, p)

}

func createPipeline(filename string, fileSize, chunkCount int) <-chan int {

	chunkSize := fileSize / chunkCount
	sortResults := []<-chan int{}
	for i := 0; i < chunkCount; i++ {
		file, err := os.Open(filename)
		if err != nil {
			panic(err)
		}

		file.Seek(int64(i*chunkSize), 0)

		pipeline.Init()
		//将分块中的数据读到管道中
		source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)

		//将排序结果收集起来
		sortResults = append(sortResults, pipeline.InMemSort(source))

	}

	return pipeline.MergeN(sortResults...)

}

func createNetworkPipeline(filename string, fileSize, chunkCount int) <-chan int {

	chunkSize := fileSize / chunkCount
	sortAddr := []string{}
	for i := 0; i < chunkCount; i++ {
		file, err := os.Open(filename)
		if err != nil {
			panic(err)
		}

		file.Seek(int64(i*chunkSize), 0)

		pipeline.Init()
		//将分块中的数据读到管道中
		source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)

		//将排序结果收集起来
		pipeline.InMemSort(source)
		addr := ":" + strconv.Itoa(9000+i)
		pipeline.NetworkSink(addr, pipeline.InMemSort(source))
		sortAddr = append(sortAddr, addr)

	}

	sortResults := []<-chan int{}
	for _, addr := range sortAddr {
		sortResults = append(sortResults, pipeline.NetworkSource(addr))
	}

	return pipeline.MergeN(sortResults...)

}
